with operations_trace_basic as (
select
waybill_no
,step
,scan_type
,network_code
,network_type
,network_name
,network_scan_time_early
,scan_time
,step_reserve
,prev_network_code
,prev_network_type
,next_network_code
,next_network_type
,step_center
,step_center_reserve
,step_entrepot
,step_entrepot_reserve
,is_sign
,named_struct(
'scan_time', scan_time,
'step_reserve', step_reserve,
'scan_type', scan_type,
'packagecode', packagecode,
'nextstationcode', nextstationcode,
'nextstation', nextstation,
'scanuser', scanuser,
'weight', weight,
'shipment_no', shipment_no,
'planned_arrival_time', planned_arrival_time,
'actual_arrival_time', actual_arrival_time,
'planned_departure_time', planned_departure_time,
'actual_departure_time', actual_departure_time
) as trace_wid
from jms_dwd.dwd_wide_rank_basic_scaninfo_tms_dt where dt >= date_add('{{ execution_date | cst_ds }}', -29) ) ,

waybill_and_step as (
select
waybill_no,                                                                  --运单号
step,                                                                        --站点
max(network_code) as network_code,                                           --站点编码
max(network_type) as network_type,                                           --站点类型
max(network_name) as network_name,                                           --站点名称
count(1) as step_all_operations_count,                                       --站点总操作次数
max(trace_wid).scan_time as scan_time_lastest,                          --站点最晚扫描时间
min(trace_wid).scan_time as scan_time_earlist,                         --站点最早扫描时间
max(trace_wid).scan_type as scan_type_lastest,                               --站点最晚扫描类型
min(trace_wid).scan_type as scan_type_earlist,                               --站点最早扫描类型
max(prev_network_code) as prev_network_code,                                 --实际上一站网点编码
max(prev_network_type) as prev_network_type,                                 --实际上一站网点类型
max(next_network_code) as next_network_code,                                 --实际下一站网点编码
max(next_network_type) as next_network_type,                                 --实际下一站网点类型
max(step_center) as step_center,                                             --中心正序
max(step_center_reserve) as step_center_reserve,                             --中心逆序
max(step_entrepot) as step_entrepot,                                         --集散正序
max(step_entrepot_reserve) as step_entrepot_reserve,                         --集散逆序
min(if(scan_type = 'taking' or scan_type ='warehousing', scan_time ,null)) as actual_taking_scantime,     --站点实际揽收时间(揽收入仓取最早)
min(if(scan_type = 'taking' or scan_type ='warehousing', trace_wid ,null)).weight as actual_taking_weight,      --站点实际揽收重量(揽收入仓取最早)
min(if(scan_type = 'taking' or scan_type ='warehousing', trace_wid ,null)).scan_type as actual_taking_scantype, --站点实际揽收类型(揽收入仓取最早)
min(if(scan_type = 'taking' or scan_type ='warehousing', trace_wid ,null)).scanuser as actual_taking_scanuser,  --站点实际揽收扫描员(揽收入仓取最早)
max(if(scan_type = 'taking',scan_time,null)) as taking_later_scantime,     --站点最晚揽收时间
min(if(scan_type = 'taking',scan_time,null)) as taking_early_scantime,    --站点最早揽收时间
min(if(scan_type = 'taking',trace_wid,null)).weight as taking_early_weight,     --运单在该站点揽收最早重量
min(if(scan_type = 'taking',trace_wid,null)).scanuser as taking_early_scanuser, --运单在该站点揽收最早扫描员
max(if(scan_type = 'warehousing',scan_time,null)) as warehousing_later_scantime,  --站点最晚入仓时间
min(if(scan_type = 'warehousing',scan_time,null)) as warehousing_early_scantime, --站点最早入仓时间
min(if(scan_type = 'warehousing',trace_wid,null)).weight as warehousing_early_weight, --站点最早入仓重量
min(if(scan_type = 'warehousing',trace_wid,null)).scanuser as warehousing_early_scanuser, --站点最早入仓扫描员
min(if(scan_type = 'bagging',scan_time,null)) as bagging_early_scantime,           --站点最早建包时间
max(if(scan_type = 'bagging',scan_time,null)) as bagging_later_scantime,            --站点最晚建包时间
max(if(scan_type = 'bagging',trace_wid,null)).packagecode as bagging_later_package_code, --站点最晚建包包号
max(if(scan_type = 'unloading',scan_time,null)) as unloading_later_scantime,  --站点最晚卸车时间
min(if(scan_type = 'unloading',scan_time,null)) as unloading_early_scantime, --站点最早卸车时间
min(if(scan_type = 'unloading',trace_wid,null)).weight as unloading_early_weight,  --站点最早卸车重量
max(if(scan_type = 'siteArrival' or scan_type = 'centerArrival',scan_time,null)) as arrival_later_scantime,     --站点最晚到件时间
min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival',scan_time,null)) as arrival_early_scantime,    --站点最早到件时间
min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival',trace_wid,null)).weight as arrival_early_weight,     --站点最早到件重量
min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival',trace_wid,null)).scanuser as arrival_early_scanuser, --站点最早到件扫描员
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).shipment_no,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).shipment_no)  as unloading_arrival_shipment_no, --站点最早卸车到件任务号
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).planned_departure_time,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).planned_departure_time)as unloading_arrival_planned_departure_time, --站点卸车到件规划发车时间
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).planned_arrival_time,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).planned_arrival_time)  as unloading_arrival_planned_arrival_time, --站点卸车到件规划到车时间
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).actual_departure_time,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).actual_departure_time)  as unloading_arrival_actual_departure_time,  --站点卸车到件实际发车时间
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).actual_arrival_time,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).actual_arrival_time)  as unloading_arrival_actual_arrival_time, --站点卸车件实际到车时间
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).nextstationcode,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).nextstationcode)  as unloading_arrival_previous_network_code, --站点卸车到件上一站站点编码
coalesce(min(if(scan_type = 'unloading',trace_wid,null)).nextstation,min(if(scan_type = 'siteArrival' or scan_type = 'centerArrival' ,trace_wid,null)).nextstation)  as unloading_arrival_previous_network_name, --站点卸车到件上一站站点名称
--min(if(scan_type = 'unloading' or scan_type = 'siteArrival' or scan_type = 'centerArrival' ,prev_network_code,null)) as unloading_arrival_previous_network_code, --站点卸车到件上一站站点编码
--min(if(scan_type = 'unloading' or scan_type = 'siteArrival' or scan_type = 'centerArrival' ,prev_network_type,null))  as unloading_arrival_previous_network_type, --站点卸车到件上一站站点类型
min(if(scan_type = 'loading',scan_time,null)) as loading_early_scantime, --站点最早装车扫描时间
max(if(scan_type = 'loading',scan_time,null)) as loading_later_scantime,  --站点最晚装车扫描时间
max(if(scan_type = 'loading',trace_wid,null)).weight as loading_later_weight,  --站点最晚装车重量
min(if(scan_type = 'siteSend' or scan_type = 'centerSend', scan_time,null)) as send_early_scantime,     --站点最早发件扫描时间
max(if(scan_type = 'siteSend' or scan_type = 'centerSend', scan_time,null)) as send_later_scantime,    --站点最晚发件扫描时间
max(if(scan_type = 'siteSend' or scan_type = 'centerSend',trace_wid,null)).weight as send_later_weight,     --站点最晚发件重量
max(if(scan_type = 'siteSend' or scan_type = 'centerSend',trace_wid,null)).scanuser as send_later_scanuser, --站点最晚发件扫描员
coalesce(max(if(scan_type = 'loading',trace_wid,null)).shipment_no,max(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).shipment_no)  as loading_send_shipment_no, --站点最晚装车发件任务号
coalesce(min(if(scan_type = 'loading',trace_wid,null)).planned_departure_time,min(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).planned_departure_time)  as loading_send_planned_departure_time,  --站点装车发件规划发车时间
coalesce(min(if(scan_type = 'loading',trace_wid,null)).planned_arrival_time,min(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).planned_arrival_time)  as loading_send_planned_arrival_time, --站点装车发件规划到车时间
coalesce(min(if(scan_type = 'loading',trace_wid,null)).actual_departure_time,min(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).actual_departure_time)  as loading_send_actual_departure_time, --站点装车发件实际发车时间
coalesce(min(if(scan_type = 'loading',trace_wid,null)).actual_arrival_time,min(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).actual_arrival_time)  as loading_send_actual_arrival_time, --站点装车发件实际到车时间
coalesce(min(if(scan_type = 'loading',trace_wid,null)).nextstationcode,min(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).nextstationcode)  as loading_send_next_network_code, --站点最晚装车发件下一站站点编码
coalesce(min(if(scan_type = 'loading',trace_wid,null)).nextstation,min(if(scan_type = 'centerSend' or scan_type = 'siteSend' ,trace_wid,null)).nextstation)  as  loading_send_next_network_name, --站点最晚装车发件下一站站点名称
--min(if(scan_type = 'loading' or scan_type = 'centerSend' or scan_type = 'siteSend' ,next_network_code,null))  as loading_send_next_network_code, --站点最晚装车发件下一站站点编码
--min(if(scan_type = 'loading' or scan_type = 'centerSend' or scan_type = 'siteSend' ,next_network_type,null))  as loading_send_next_network_type, --站点最晚装车发件下一站站点类型
min(if(scan_type = 'deliver',scan_time,null)) as deliver_early_scantime,     --站点最早派件扫描时间
max(if(scan_type = 'deliver',scan_time,null)) as deliver_later_scantime,      --站点最晚派件扫描时间
max(if(scan_type = 'deliver',trace_wid,null)).weight as deliver_later_weight,      --站点最晚派件重量
max(if(scan_type = 'deliver',trace_wid,null)).scanuser as deliver_later_scanuser,  --站点最晚派件扫描员
min(if(scan_type = 'sign' or scan_type = 'outStore',scan_time,null)) as actual_sign_scantime,     --站点实际签收时间(签收出库取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore',trace_wid,null)).weight as actual_sign_weight,      --站点实际签收重量(签收出库取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore',trace_wid,null)).scan_type as actual_sign_scantype, --站点实际签收类型(签收出库取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore',trace_wid,null)).scanuser as actual_sign_scanuser,  --站点实际签收扫描员(签收出库取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore' or scan_type = 'inStore',scan_time,null)) as aging_sign_scantime, -- 站点时效签收时间(签收,出库,入库,代理点收入取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore' or scan_type = 'inStore',trace_wid,null)).weight as aging_sign_weight,  -- 站点时效签收重量(签收,出库,入库,代理点收入取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore' or scan_type = 'inStore',trace_wid,null)).scan_type as aging_sign_scantype, -- 站点时效签收类型(签收,出库,入库,代理点收入取最早)
min(if(scan_type = 'sign' or scan_type = 'outStore' or scan_type = 'inStore',trace_wid,null)).scanuser as aging_sign_scanuser, -- 站点时效签收扫描员(签收,出库,入库,代理点收入取最早)
date_format(max(scan_time),'yyyy-MM-dd') as dt
from operations_trace_basic group by waybill_no, step
)
insert overwrite table jms_dwd.dwd_s03_whole_operations_trace_step_dt partition(dt) select * from waybill_and_step
 distribute by dt, abs(hash(waybill_no)) % 100 ;

